package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator

object Demo23Acc {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("app")
    conf.setMaster("local")

    val sc = new SparkContext(conf)

    val studentsRDD: RDD[String] = sc.textFile("data/students.txt")

    /**
     * 累加器: 用于全局累加的一个变量
     * 累加过程
     * 1、在Executor对累加器进行累加
     * 2、将多个task的累加加过拉取到Driver段进行汇总
     */

    //1、在Driver段定义累加器
    val num: LongAccumulator = sc.longAccumulator

    val ages: RDD[Double] = studentsRDD.map(line => {
      //2、在算子内对累加器进行累加
      num.add(1)

      val age: Double = line.split(",")(2).toDouble
      age
    })

    //读取累加结果必须在Action算子之后读取
    println("value:" + num.value)

    //计算总的年龄,sum:也是一个action赛中你
    val sumAge: Double = ages.sum()

    //3、在Driver段获取累加器的值（num.value）
    val avgAge: Double = sumAge / num.value

    /**
     * 使用了两个action算子，会产生两个job, 前面的RDD会被计算两次，性能低
     */
    //val avgAge: Double = ages.sum() / ages.count()

    println(avgAge)


  }

}
